package bootsxt;

import bootsxt.utils.ImageCompressor;
import bootsxt.utils.Kafka;
import bootsxt.utils.MsgType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.yeauty.annotation.EnableWebSocket;

import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.OutputStream;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Spring Boot 应用程序的启动类。
 */
@EnableScheduling
@EnableWebSocket
@SpringBootApplication
@RestController
public class CameraStreamingApplication {

    private static final Logger log = LoggerFactory.getLogger(CameraStreamingApplication.class);

    public static void main(String[] args) {


        SpringApplication.run(CameraStreamingApplication.class, args);
    }

    // 每秒执行一次
//    @Scheduled(fixedRate = 1000)
//    public void performTask() {

//    }
@GetMapping(value = "/stream", produces = "multipart/x-mixed-replace; boundary=frame")
public void streamVideo(HttpServletResponse response, @RequestParam(defaultValue = "4") Integer quality, @RequestParam String id) throws IOException {
    response.setContentType("multipart/x-mixed-replace; boundary=frame");
    // 设置缓存头部，确保缓存不会过期，且在一定时间内不重新加载
    response.setHeader("Cache-Control", "public, max-age=60");  // 60秒缓存，视实际需求调整
    response.setHeader("Pragma", "cache");
    response.setHeader("Expires", "60");
    try (OutputStream out = response.getOutputStream()) {
        writeFrameToResponse(out, id,quality);
    } catch (IOException | InterruptedException e) {
        System.err.println("客户端连接中断或其他I/O异常：" + e.getMessage());
    }
}

    @GetMapping(value = "/isUpload")
    public int isUpload(@RequestParam String id) throws IOException {
       return Kafka.topic(id, MsgType.IMAGE.name(), byte[].class).getScribes().size();
    }

    private void writeFrameToResponse(OutputStream out, String id, float quality) throws IOException, InterruptedException {
        Kafka.Topic<byte[]> topic = Kafka.topic(id, MsgType.IMAGE.name(), byte[].class);
        String uid = UUID.randomUUID().toString();
        try {
            topic.scribes(bytes -> {
                try {
                    out.write("--frame\r\n".getBytes());
                    out.write("Content-Type: image/jpeg\r\n\r\n".getBytes());
                    out.write(ImageCompressor.scaleImage(bytes, quality));
                    out.write("\r\n".getBytes());
                    out.flush();
                }catch (Throwable e) {
                    topic.unsubscribe(uid); // 取消订阅
                }
                return null;
            },uid);
            while (topic.isScribe(uid)){
                boolean b = topic.pingMsg(30 * 1000);
                if (!b){
                    log.info("订阅 topic：{} 没有消息了",topic);
                    break;
                }
            }
        }finally {
            topic.unsubscribe(uid);
        }
    }



}
